package org.devgframwork.datasync.es56.executors;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.GsonBuilder;

import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.core.*;
import lombok.extern.slf4j.Slf4j;
import org.devgframwork.datasync.es56.entity.ElasticSearchConfigPO;
import org.devgframwork.datasync.es56.entity.TargetConfigPO;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class Es56Executor {

    public enum EsExecutorType {

        INDEX("i"), UPDATE("u"), DELETE("d");

        EsExecutorType(String typeName) {
            this.typeName = typeName;
        }

        private String typeName;

        public String getTypeName() {
            return typeName;
        }

    }

    public static final Map<String, JestClient> es56Map = new ConcurrentHashMap<>();

    public static boolean indexDocBulk(TargetConfigPO targetConfigPO, List<Map<String, Object>> list) {
        Integer retryTimes = targetConfigPO.getRetryTimes();
        boolean bool = updateIndexDocBulk(targetConfigPO, list);
        if (!bool && retryTimes > 0) {
            targetConfigPO.setRetryTimes(retryTimes - 1);
            bool = indexDocBulk(targetConfigPO, list);
        }
        return bool;
    }

    public static boolean updateIndexDocBulk(TargetConfigPO targetConfigPO, List<Map<String, Object>> list) {
        boolean boll = false;
        try {
            String targetContent = targetConfigPO.getTargetContent();
            if (null == targetContent || targetContent.trim().equals("")) {
                log.error("targetContent cannot be null");
                return false;
            }
            String targetKey = targetConfigPO.getTargetKey();
            JestClient jestClient = es56Map.get(targetKey);
            if (null == jestClient) {
                log.error("jestClient cannot be null");
                return false;
            }
            String[] indexInfo = targetContent.split("\\/");
            String doType = EsExecutorType.INDEX.getTypeName();
            if (indexInfo.length >= 3) {
                doType = indexInfo[2];
            }
            String id = targetConfigPO.getTargetId();
            Bulk.Builder bulk = new Bulk.Builder().defaultIndex(indexInfo[0]).defaultType(indexInfo[1]);
            for (Map<String, Object> data : list) {
                if (doType.equals(EsExecutorType.UPDATE.getTypeName())) {
                    String json = JSONObject.toJSONString(data);
                    String builderStr = "{\"doc\":" + json + ",\"doc_as_upsert\":false}";
                    log.info("upsetById builderStr:{}", builderStr);
                    Update update = new Update.Builder(builderStr).id(data.get(id).toString()).build();
                    bulk.addAction(update);
                    continue;
                }
                Index index;
                if (null != id && !id.trim().equals("")) {
                    index = new Index.Builder(data).id(data.get(id).toString()).build();
                } else {
                    //如未设置索引唯一id值，则唯一id会默认生成，索引操作为添加操作
                    index = new Index.Builder(data).build();
                }
                bulk.addAction(index);
            }
            BulkResult br = jestClient.execute(bulk.build());
            boll = br.isSucceeded();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return boll;
    }

    public static void initEsConfig(ElasticSearchConfigPO elasticSearchConfigPO) {
        synchronized (Es56Executor.class) {
            JestClientFactory factory = new JestClientFactory();
            String serverUri = elasticSearchConfigPO.getSchemeName() + "://" + elasticSearchConfigPO.getHostname() + ":" + elasticSearchConfigPO.getPort();
            factory.setHttpClientConfig(new HttpClientConfig.Builder(serverUri)
                    .gson(new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss").create())
                    .multiThreaded(true)
                    .connTimeout(elasticSearchConfigPO.getConnectionTimeOut())
                    .readTimeout(elasticSearchConfigPO.getSocketTimeOut())
                    .defaultMaxTotalConnectionPerRoute(elasticSearchConfigPO.getMaxConnectPerRoute())
                    .maxTotalConnection(elasticSearchConfigPO.getMaxConnectTotal())
                    .build());
            JestClient jestClient = factory.getObject();
            es56Map.put(elasticSearchConfigPO.getKeyName(), jestClient);
            log.info("init EsSearch jestClient success ... ");
        }
    }
}
